/*


Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controllers

import (
	"context"
	"fmt"
	"github.com/go-logr/logr"
	"github.com/sirupsen/logrus"
	ckev1alpha1 "iop.inspur.com/api/v1alpha1"
	"iop.inspur.com/constants"
	"iop.inspur.com/util"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"os"
	"path/filepath"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sort"
	"strings"
	"time"
)

const (
	CkeadmContainerName    = "ckeamd-container"
	ComponentContainerName = "component-container"
	ComponentFinalizerName = "finalizer.stable.component.cke.io"

	ComponentInstallJobAppLabel   = "cke-component-install-job"
	ComponentScaleupJobAppLabel   = "cke-component-scaleup-job"
	ComponentScaledownJobAppLabel = "cke-component-scaledown-job"
	ComponentUpdateJobAppLabel    = "cke-component-update-job"
	ComponentUpgradeJobAppLabel   = "cke-component-upgrade-job"
	ComponentTerminateJobAppLabel = "cke-component-terminate-job"

	ComponentTerminatingStatus     = "Terminating"
	ComponentInstallingStatus      = "Installing"
	ComponentUpgradingStatus       = "Upgrading"
	ComponentUpdatingStatus        = "Updating"
	ComponentScalingUpStatus       = "ScalingUp"
	ComponentScalingDownStatus     = "ScalingDown"
	ComponentRunningStatus         = "Running"
	ComponentValidateFailedStatus  = "ValidateFailed"
	ComponentInstallFailedStatus   = "InstallFailed"
	ComponentUpgradeFailedStatus   = "UpgradeFailed"
	ComponentUpdateFailedStatus    = "UpdateFailed"
	ComponentTerminateFailedStatus = "TerminateFailed"
	ComponentInstalled             = "Installed"

	BaseComponentLabel          = "baseComponent"
	UnBaseComponentDepend       = "cke-controller"
	CkeControllerServiceName    = "svc-ckeadm"
	CkeNodeScaleupRole          = "kube-scale"
	CkeNodeScaleDownRole        = "kube-scaledown"
	CkeComponentScriptImageRepo = "library/cke/components/"
	CkeadmImageRepo             = "library/cke/cke-admin"
	ScriptPathInScriptImage     = "/template/*"
	CkeadmCkePatchPath          = "/opt/patch-cr"
	CkeComponentLockPath        = "/etc/kubernetes/cke-controller/lock"
	CkeControllerCkePatchPath   = "/etc/kubernetes/cke-controller/patch-cr"
)

// CkeComponentReconciler reconciles a CkeComponent object
type CkeComponentReconciler struct {
	client.Client
	Log    logr.Logger
	Scheme *runtime.Scheme
}

// +kubebuilder:rbac:groups=cke.inspur.com,resources=ckecomponents,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cke.inspur.com,resources=ckecomponents/status,verbs=get;update;patch

//todo: 关键组件升级，卸载权限开关
func (r *CkeComponentReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, ************** ", req.NamespacedName)
	var requeue = false
	var err error

	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, get CkeComponent.", req.NamespacedName)
	//get ckecomponent
	ctx := context.Background()
	ckecomponent := &ckev1alpha1.CkeComponent{}
	if err = r.Get(ctx, req.NamespacedName, ckecomponent); err != nil {
		if errors.IsNotFound(err) {
			logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, CkeComponent not found.", req.NamespacedName)
			return ctrl.Result{}, nil
		}
		logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, failed to get CkeComponent. error=%s", req.NamespacedName, err.Error())
		return ctrl.Result{}, err
	}
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, success to get CkeComponent(version=%s)", req.NamespacedName, ckecomponent.Spec.Version)

	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, get CkeComponentList.", req.NamespacedName)
	//get ckecomponentes
	ckecomponents := &ckev1alpha1.CkeComponentList{}
	if err = r.List(ctx, ckecomponents); err != nil {
		logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, failed to get CkeComponentList. error=%s", req.NamespacedName, err.Error())
		return ctrl.Result{}, err
	}

	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, get CkeNodeList.", req.NamespacedName)

	//get ckenodes
	ckenodes := &ckev1alpha1.CkeNodeList{}
	if err = r.List(ctx, ckenodes); err != nil {
		if errors.IsNotFound(err) {
			requeue = true
			logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, CkeNodeList not found.", req.NamespacedName)
			return ResultIfRequeue(requeue), nil
		}
		logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, failed to get CkeNodeList. error=%s", req.NamespacedName, err.Error())
		return ctrl.Result{}, err
	}
	nodes := make([]string, 0)
	scaleupNodes := make([]string, 0)
	rolechangeNode := make([]string, 0)
	for _, ckenode := range ckenodes.Items {
		nodes = append(nodes, ckenode.Name)
		if len(ckenode.Status.ScaleupComponents) != 0 || len(ckenode.Status.ActiveScaleupComponent) != 0 || len(ckenode.Status.CompletedScaleupComponents) != 0 {
			scaleupNodes = AppendIfNotExist(scaleupNodes, ckenode.Name)
		}
		if ckenode.Status.CkeRole != getNodeRole(ckenode.Spec.Roles) {
			rolechangeNode = AppendIfNotExist(rolechangeNode, ckenode.Name)
		}
	}
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, success to get CkeNodeList. allNode=%s", req.NamespacedName, strings.Join(nodes, ", "))

	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, get CkeClusterList.", req.NamespacedName)
	//get ckeclusters
	ckeclusters := &ckev1alpha1.CkeClusterList{}
	if err = r.List(ctx, ckeclusters); err != nil {
		logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, failed to get CkeClusterList. error=%s", req.NamespacedName, err.Error())
		return ctrl.Result{}, err
	}
	clusters := make([]string, 0)
	for _, ckecluster := range ckeclusters.Items {
		clusters = append(clusters, ckecluster.Name)
	}
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, success to get CkeClusterList. allCluster=%s", req.NamespacedName, strings.Join(clusters, ", "))

	if len(ckeclusters.Items) > 1 || len(ckeclusters.Items) < 1 {
		logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, CKE Cluster count is not 1! count: %d", req.NamespacedName, len(ckeclusters.Items))
		return ctrl.Result{}, nil
	}
	var Cluster *ckev1alpha1.CkeCluster
	Cluster = &ckeclusters.Items[0]
	if Cluster == nil {
		return ctrl.Result{}, fmt.Errorf("CkeComponent Reconcile, CkeComponent=%s, not find ckecluster", req.NamespacedName)
	}

	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, get svc-ckeadm Service.", req.NamespacedName)
	//get service
	serviceNamespacedName := client.ObjectKey{
		Name:      CkeControllerServiceName,
		Namespace: req.Namespace,
	}
	service := &corev1.Service{}
	if err = r.Get(ctx, serviceNamespacedName, service); err != nil {
		logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, failed to get svc-ckeadm service. error=%s", req.NamespacedName, err.Error())
		return ctrl.Result{}, err
	}

	//add finalizers to ckecomponent
	if ckecomponent.DeletionTimestamp.IsZero() {
		if !ContainsStrings(ckecomponent.Finalizers, ComponentFinalizerName) {
			ckecomponent.Finalizers = append(ckecomponent.Finalizers, ComponentFinalizerName)
			logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, add finalizer. finalizer=%s", req.NamespacedName, ComponentFinalizerName)
			if err := r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, failed to add finalizers. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
		}
	}

	//terminate ckecomponent
	if ckecomponent.Status.State == ComponentTerminatingStatus {
		logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, terminate CkeComponent.", req.NamespacedName)
		requeue, err = r.TerminateComponent(ctx, service, ckecomponent, ckenodes.Items, Cluster)
		if err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, failed to terminate CkeComponent. error=%s", req.NamespacedName, err.Error())
			return ctrl.Result{}, err
		}
		return ResultIfRequeue(requeue), nil
	}
	if !ckecomponent.DeletionTimestamp.IsZero() && ContainsStrings(ckecomponent.Finalizers, ComponentFinalizerName) {
		if ckecomponent.Status.State == ComponentTerminateFailedStatus || ckecomponent.Status.State == ComponentInstallFailedStatus {
			logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, InstallComponent or TerminateComponent already failed, delete directly", req.NamespacedName)
			ckecomponent.Finalizers = RemoveStrings(ckecomponent.Finalizers, ComponentFinalizerName)
			logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, delete CkeComponent by remove finalizer.stable.component.cke.io.", req.NamespacedName)
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, failed to delete CkeComponent by remove finalizer.stable.component.cke.io. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
			return ctrl.Result{}, err
		} else if ckecomponent.Status.State == ComponentValidateFailedStatus && ckecomponent.Status.LastState == "" {
			logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, delete CkeComponent by remove finalizer.stable.component.cke.io.", req.NamespacedName)
			ckecomponent.Finalizers = RemoveStrings(ckecomponent.Finalizers, ComponentFinalizerName)
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			return ctrl.Result{}, err
		} else {
			logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, try to delete residual jobs.", req.NamespacedName)
			var CkeJobs = &batchv1.JobList{}
			var JobSelector = map[string]string{
				"componentName": ckecomponent.Name,
				"clusterID":     Cluster.Name,
			}
			if err = r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s job list error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			if len(CkeJobs.Items) > 0 {
				for _, componentJob := range CkeJobs.Items {
					if err = r.Delete(ctx, &componentJob, &constants.BackGroundDeleted); err != nil {
						logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s job delete error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
						return ctrl.Result{}, err
					}
				}
			}
			requeue, err = r.TerminateComponent(ctx, service, ckecomponent, ckenodes.Items, Cluster)
			if err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s TerminateComponent error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
		}
		return ResultIfRequeue(requeue), nil
	}

	//validate ckecomponent
	if err = r.ValidateCkeComponent(ckecomponent, Cluster.Name); err != nil {
		logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s validateCkeComponent error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
		if ckecomponent.Status.State != ComponentValidateFailedStatus {
			ckecomponent.Status.LastState = ckecomponent.Status.State
		}
		ckecomponent.Status.State = ComponentValidateFailedStatus
		ckecomponent.Status.Reason = "validateCkeComponent failed"
		ckecomponent.Status.Message = err.Error()
		if err = r.Update(ctx, ckecomponent); err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
			return ctrl.Result{}, err
		}
		return ctrl.Result{}, nil
	}
	if ckecomponent.Status.State == ComponentValidateFailedStatus {
		ckecomponent.Status.State = ckecomponent.Status.LastState
		ckecomponent.Status.LastState = ""
		ckecomponent.Status.Reason = ""
		ckecomponent.Status.Message = ""
		if err = r.Update(ctx, ckecomponent); err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
			return ctrl.Result{}, err
		}
	}

	//set baseComponent label
	changeLabels := false
	labels := make(map[string]string)
	if len(ckecomponent.Labels) == 0 {
		labels[BaseComponentLabel] = "false"
		ckecomponent.Labels = labels
		changeLabels = true
	} else if ckecomponent.Labels[BaseComponentLabel] != "true" && ckecomponent.Labels[BaseComponentLabel] != "false" {
		ckecomponent.Labels[BaseComponentLabel] = "false"
		changeLabels = true
	}
	if changeLabels {
		if err = r.Update(ctx, ckecomponent); err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update label error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
			return ctrl.Result{}, err
		}
		logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, set label baseComponent=%s", req.NamespacedName, ckecomponent.Labels[BaseComponentLabel])
	}

	//set unBaseComponent depend
	if ckecomponent.Labels[BaseComponentLabel] == "false" && !ContainsStrings(ckecomponent.Spec.Depend, UnBaseComponentDepend) {
		ckecomponent.Spec.Depend = append(ckecomponent.Spec.Depend, UnBaseComponentDepend)
		if err = r.Update(ctx, ckecomponent); err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update depend error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
			return ctrl.Result{}, err
		}
		logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, set unBaseComponent depend %s", req.NamespacedName, UnBaseComponentDepend)
	}

	//set ckecomponent imageTag
	if ckecomponent.Spec.ImageTag == "" {
		ckecomponent.Spec.ImageTag = ckecomponent.Spec.Version
		if err = r.Update(ctx, ckecomponent); err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update imageTag error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
			return ctrl.Result{}, err
		}
		logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, set imageTag %s", req.NamespacedName, ckecomponent.Spec.ImageTag)
	}

	//set ckecomponent nodeSelector
	if len(ckecomponent.Spec.Nodes) == 0 {
		if ckecomponent.Status.State == constants.EmptyString || ckecomponent.Status.State == ComponentInstalled {
			ckecomponent.Status.NodeSelector = ckecomponent.Spec.NodeSelector
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
		} else {
			if !SliceEqual(ckecomponent.Spec.NodeSelector, ckecomponent.Status.NodeSelector) {
				ckecomponent.Status.Reason = "Not allowed"
				ckecomponent.Status.Message = "Not allowed to change component's nodeSelector"
				ckecomponent.Spec.NodeSelector = ckecomponent.Status.NodeSelector
				if err = r.Update(ctx, ckecomponent); err != nil {
					logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
					return ctrl.Result{}, err
				}
			}
		}
	}

	//install ckecomponent
	if ckecomponent.Status.State == ComponentInstalled {
		logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, component already installed, update status directly", req.NamespacedName)
		ckecomponent.Status.State = ComponentRunningStatus
		ckecomponent.Status.Version = ckecomponent.Spec.Version
		ckecomponent.Status.ImageTag = ckecomponent.Spec.ImageTag
		ckecomponent.Status.Parameters = ckecomponent.Spec.Parameters
		if len(ckecomponent.Status.Nodes) == 0 {
			nodes, _, err := r.GetMatchNodes(ctx, ckecomponent.Spec.Nodes, ckecomponent.Spec.NodeSelector, ckecomponent.Spec.Replicas)
			if err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s GetMatchNodes error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			ckecomponent.Status.Nodes = nodes
		}
		if err = r.Update(ctx, ckecomponent); err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
			return ctrl.Result{}, err
		}
		return ctrl.Result{}, nil
	}
	if ckecomponent.Status.State == constants.EmptyString || ckecomponent.Status.State == ComponentInstallingStatus {
		requeue, err = r.InstallComponent(ctx, service, ckecomponent, ckecomponents, ckenodes.Items, Cluster)
		if err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s InstallComponent error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
			return ctrl.Result{}, err
		}
		requeue = true
		return ResultIfRequeue(requeue), nil
	}

	//not allowed to downgrade
	if ckecomponent.Spec.Version != ckecomponent.Status.Version {
		versionCompare := util.CompareStrVer(ckecomponent.Status.Version, ckecomponent.Spec.Version)
		if versionCompare == util.VersionBig {
			ckecomponent.Status.Message = fmt.Sprintf("Not allowed to downgrade, from version: %s; to version: %s", ckecomponent.Status.Version, ckecomponent.Spec.Version)
			ckecomponent.Spec.Version = ckecomponent.Status.Version
			ckecomponent.Spec.ImageTag = ckecomponent.Status.ImageTag
			ckecomponent.Spec.Parameters = ckecomponent.Status.Parameters
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			requeue = true
			return ResultIfRequeue(requeue), nil
		}
	}
	if ckecomponent.Spec.ImageTag != ckecomponent.Status.ImageTag {
		imageTagCompare := util.CompareStrVer(ckecomponent.Status.ImageTag, ckecomponent.Spec.ImageTag)
		if imageTagCompare == util.VersionBig {
			ckecomponent.Status.Message = fmt.Sprintf("Not allowed to downgrade, from imageTag: %s; to imageTag: %s", ckecomponent.Status.ImageTag, ckecomponent.Spec.ImageTag)
			ckecomponent.Spec.ImageTag = ckecomponent.Status.ImageTag
			ckecomponent.Spec.Parameters = ckecomponent.Status.Parameters
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s Update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			requeue = true
			return ResultIfRequeue(requeue), nil
		}
	}

	//upgrade or update ckecomponent
	if ckecomponent.Spec.Version != ckecomponent.Status.Version {
		if ckecomponent.Status.State == ComponentRunningStatus || ckecomponent.Status.State == ComponentUpgradingStatus || ckecomponent.Status.State == ComponentUpgradeFailedStatus {
			requeue, err = r.UpgradeComponent(ctx, service, ckecomponent, ckecomponents, ckenodes.Items, Cluster)
			if err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s UpgradeComponent error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			requeue = true
			return ResultIfRequeue(requeue), nil
		}
	} else if ckecomponent.Spec.ImageTag != ckecomponent.Status.ImageTag || IsParametersChanged(ckecomponent) {
		if ckecomponent.Status.State == ComponentRunningStatus || ckecomponent.Status.State == ComponentUpdatingStatus || ckecomponent.Status.State == ComponentUpdateFailedStatus {
			requeue, err = r.UpdateComponent(ctx, service, ckecomponent, ckecomponents, ckenodes.Items, Cluster)
			if err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s UpdateComponent error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			requeue = true
			return ResultIfRequeue(requeue), nil
		}
	}

	//scaleup ckecomponent
	if len(ckecomponent.Status.ScalingUpPending) > 0 {
		if ckecomponent.Status.State == ComponentRunningStatus || ckecomponent.Status.State == ComponentScalingUpStatus {
			requeue, err = r.ScaleupComponent(ctx, service, ckecomponent, ckecomponents, ckenodes.Items, Cluster)
			if err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s ScaleupComponent error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			requeue = true
			return ResultIfRequeue(requeue), nil
		}
	}

	//scaledown ckecomponent
	if len(ckecomponent.Status.ScalingDownPending) > 0 {
		if ckecomponent.Status.State == ComponentRunningStatus || ckecomponent.Status.State == ComponentScalingDownStatus {
			requeue, err = r.ScaledownComponent(ctx, service, ckecomponent, ckenodes.Items, Cluster)
			if err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s ScaledownComponent error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			requeue = true
			return ResultIfRequeue(requeue), nil
		}
	}

	//check if .status.nodes are still matched
	if ckecomponent.Status.State == ComponentRunningStatus {
		nodes, notEnough, err := r.GetMatchNodes(ctx, ckecomponent.Spec.Nodes, ckecomponent.Spec.NodeSelector, ckecomponent.Spec.Replicas)
		if err != nil {
			logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s GetMatchNodes error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
			return ctrl.Result{}, err
		}
		if notEnough {
			ckecomponent.Status.Reason = fmt.Sprintf("the replicas of component %s is wrong", ckecomponent.Name)
			ckecomponent.Status.Message = fmt.Sprintf("MatchNodes %s are not enough for replicas %v.", strings.Join(nodes, ","), ckecomponent.Spec.Replicas)
			ckecomponent.Spec.Replicas = len(ckecomponent.Status.Nodes)
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s update error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
				return ctrl.Result{}, err
			}
			requeue = true
			return ResultIfRequeue(requeue), nil
		}
		if SliceEqual(ckecomponent.Status.Nodes, nodes) || len(ckecomponent.Status.Nodes) == ckecomponent.Spec.Replicas {
			if !sort.StringsAreSorted(ckecomponent.Status.Nodes) {
				sort.Strings(ckecomponent.Status.Nodes)
				if err := r.Update(ctx, ckecomponent); err != nil {
					logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s update sorted .status.nodes error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
					return ctrl.Result{}, err
				}
			}
		} else {
			scalingup := RemoveStrings(RemoveStrings(RemoveStrings(difference(nodes, ckecomponent.Status.Nodes), ckecomponent.Status.ScalingUpFailed...), scaleupNodes...), rolechangeNode...)
			scalingdown := RemoveStrings(RemoveStrings(difference(ckecomponent.Status.Nodes, nodes), ckecomponent.Status.ScalingDownFailed...), rolechangeNode...)
			logrus.Infof("matchNodes: %s, scalingup: %s, scalingdown: %s", strings.Join(nodes, ","), strings.Join(scalingup, ","), strings.Join(scalingdown, ","))
			if len(scalingup) > 0 || len(scalingdown) > 0 {
				ckecomponent.Status.ScalingUpPending = AppendIfNotExist(ckecomponent.Status.ScalingUpPending, scalingup...)
				ckecomponent.Status.ScalingDownPending = AppendIfNotExist(ckecomponent.Status.ScalingDownPending, scalingdown...)
				if err := r.Update(ctx, ckecomponent); err != nil {
					logrus.Errorf("CkeComponent Reconcile, CkeComponent=%s, %s/%s update ScalingUpPending and ScalingDownPending error! err: %s", req.NamespacedName, req.Namespace, ckecomponent.Name, err.Error())
					return ctrl.Result{}, err
				}
				requeue = true
				return ResultIfRequeue(requeue), nil
			}
		}
	}

	return ResultIfRequeue(requeue), nil
}

func (r *CkeComponentReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&ckev1alpha1.CkeComponent{}).
		Complete(r)
}

func (r *CkeComponentReconciler) ValidateCkeComponent(ckecomponent *ckev1alpha1.CkeComponent, clusterName string) error {
	errs := make([]string, 0)
	if ckecomponent.Spec.Version == "" {
		errs = append(errs, " .spec.version is not configured")
	}
	if ckecomponent.Spec.ClusterName != clusterName {
		errs = append(errs, " .spec.clusterName is not "+clusterName)
	}
	if len(ckecomponent.Spec.Nodes) > 0 && len(ckecomponent.Spec.NodeSelector) > 0 {
		errs = append(errs, " .spec.nodes and .spec.nodeSelector are configured together")
	}
	if len(ckecomponent.Spec.Nodes) > 0 && ckecomponent.Spec.Replicas != 0 {
		errs = append(errs, " .spec.nodes and .spec.replicas are configured together")
	}
	if len(errs) != 0 {
		return fmt.Errorf(strings.Join(errs, ";"))
	}
	return nil
}

func (r *CkeComponentReconciler) GetMatchNodes(ctx context.Context, nodes, nodeSelector []string, replicas int) ([]string, bool, error) {
	notEnough := false
	matchNodes := make([]string, 0)
	ckenodes := &ckev1alpha1.CkeNodeList{}
	if err := r.List(ctx, ckenodes); err != nil {
		return nil, notEnough, err
	}
	if len(nodes) > 0 {
		matchNodes = nodes
		return matchNodes, notEnough, nil
	} else if len(nodeSelector) > 0 {
		for _, ckenode := range ckenodes.Items {
			if ckenode.Status.CkeRole != getNodeRole(ckenode.Spec.Roles) && getNodeRole(ckenode.Spec.Roles) == constants.CkeNodeRoleWorker && ContainsStrings(nodeSelector, constants.NodeMasterLabel) {
				continue
			}
			if ckenode.DeletionTimestamp.IsZero() && ContainsStrings(ckenode.Status.Labels, nodeSelector...) {
				matchNodes = append(matchNodes, ckenode.Name)
			}
		}
	} else {
		for _, ckenode := range ckenodes.Items {
			if ckenode.DeletionTimestamp.IsZero() {
				matchNodes = append(matchNodes, ckenode.Name)
			}
		}
	}
	sort.Strings(matchNodes)
	if replicas != 0 {
		if replicas <= len(matchNodes) {
			matchNodes = matchNodes[:replicas]
		} else {
			notEnough = true
			return matchNodes, notEnough, nil
		}
	}
	return matchNodes, notEnough, nil
}

func (r *CkeComponentReconciler) TerminateComponent(ctx context.Context, service *corev1.Service, ckecomponent *ckev1alpha1.CkeComponent, ckenodes []ckev1alpha1.CkeNode, ckecluster *ckev1alpha1.CkeCluster) (bool, error) {
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, TerminateComponent start")
	var requeue = false
	var err error

	if strings.Contains(ckecomponent.Spec.UnsupportAction, "delete") {
		ckecomponent.Finalizers = RemoveStrings(ckecomponent.Finalizers, ComponentFinalizerName)
		if err = r.Update(ctx, ckecomponent); err != nil {
			logrus.Errorf("Terminate Component, update component error!")
			return requeue, err
		}
		return requeue, err
	}

	if ckecomponent.Status.State != ComponentTerminatingStatus {
		ckecomponent.Status.LastState = ckecomponent.Status.State
		ckecomponent.Status.State = ComponentTerminatingStatus
		ckecomponent.Status.Reason = ""
		ckecomponent.Status.Message = ""
		if err = r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
	}

	var CkeJobs = &batchv1.JobList{}
	var JobSelector = map[string]string{
		"app":           ComponentTerminateJobAppLabel,
		"componentName": ckecomponent.Name,
		"clusterID":     ckecluster.Name,
	}
	if err = r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
		return requeue, err
	}

	if len(CkeJobs.Items) == 0 {
		if err = r.Create(ctx, newComponentTerminateJob(service.Spec.Ports[0].NodePort, ckecomponent, ckenodes, ckecluster)); err != nil {
			logrus.Errorf("Terminate Component, newComponentTerminateJob error!")
			return requeue, err
		}
		logrus.Infof("success create job for terminate component(%s)", ckecomponent.Name)
		if err = r.Update(ctx, ckecomponent); err != nil {
			logrus.Errorf("Terminate Component, update component status error!")
			return requeue, err
		}
	} else {
		if JobCompleted(&CkeJobs.Items[0]) {
			if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
				logrus.Errorf("Terminate Component, delete job error!")
				return requeue, err
			}
			ckecomponent.Finalizers = RemoveStrings(ckecomponent.Finalizers, ComponentFinalizerName)
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("Terminate Component, update component error!")
				return requeue, err
			}
			return requeue, err
		}
		if JobFailed(&CkeJobs.Items[0]) {
			if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
				logrus.Errorf("Terminate Component, delete job error!")
				return requeue, err
			}
			ckecomponent.Status.State = ComponentTerminateFailedStatus
			ckecomponent.Status.Reason = fmt.Sprintf("terminate component %s failed", ckecomponent.Name)
			ckecomponent.Status.Message = fmt.Sprintf("See log in /var/log/%s/%s/delete for details.", ckecluster.Name, ckecomponent.Name)
			ckecomponent.Finalizers = RemoveStrings(ckecomponent.Finalizers, ComponentFinalizerName)
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("Terminate Component, update component status error!")
				return requeue, err
			}
			namespacedName := types.NamespacedName{
				Namespace: ckecomponent.Namespace,
				Name:      ckecomponent.Name,
			}
			for {
				component := &ckev1alpha1.CkeComponent{}
				if err = r.Get(ctx, namespacedName, component); err != nil {
					if errors.IsNotFound(err) {
						break
					}
					return requeue, err
				}
				time.Sleep(time.Second)
			}
			if err = r.Create(ctx, ckecomponent); err != nil {
				logrus.Errorf("Terminate Component, update component status error!")
				return requeue, err
			}
			return requeue, nil
		}
	}

	requeue = true
	return requeue, err
}

func (r *CkeComponentReconciler) InstallComponent(ctx context.Context, service *corev1.Service, ckecomponent *ckev1alpha1.CkeComponent, ckecomponents *ckev1alpha1.CkeComponentList, ckenodes []ckev1alpha1.CkeNode, ckecluster *ckev1alpha1.CkeCluster) (bool, error) {
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, InstallComponent start")
	var requeue = false
	var err error

	nodes, notEnough, err := r.GetMatchNodes(ctx, ckecomponent.Spec.Nodes, ckecomponent.Spec.NodeSelector, ckecomponent.Spec.Replicas)
	if err != nil {
		logrus.Errorf("Install Component, GetMatchNodes error!")
		return requeue, err
	}
	if notEnough {
		ckecomponent.Status.State = ComponentInstallFailedStatus
		ckecomponent.Status.Reason = fmt.Sprintf("install component %s failed", ckecomponent.Name)
		ckecomponent.Status.Message = fmt.Sprintf("MatchNodes %s are not enough for replicas %v.", strings.Join(nodes, ","), ckecomponent.Spec.Replicas)
		if err = r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
		return requeue, err
	}

	if strings.Contains(ckecomponent.Spec.UnsupportAction, "create") {
		ckecomponent.Status.State = ComponentRunningStatus
		ckecomponent.Status.Version = ckecomponent.Spec.Version
		ckecomponent.Status.ImageTag = ckecomponent.Spec.ImageTag
		ckecomponent.Status.Parameters = ckecomponent.Spec.Parameters
		ckecomponent.Status.Nodes = nodes
		if err = r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
		return requeue, err
	}

	if ckecomponent.Status.State != ComponentInstallingStatus {
		ckecomponent.Status.State = ComponentInstallingStatus
		ckecomponent.Status.Reason = ""
		ckecomponent.Status.Message = ""
		if err := r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
	}

	if !IsDependComponentRunning(ckecomponent, ckecomponents) {
		if ckecomponent.Status.Reason == "" {
			ckecomponent.Status.Reason = fmt.Sprintf("some depend ckecomponents are not running")
			ckecomponent.Status.Message = fmt.Sprintf("wait for depend ckecomponents running")
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}
	} else {
		if ckecomponent.Spec.Lock != "" {
			locked := false
			err = filepath.Walk(CkeComponentLockPath, func(path string, info os.FileInfo, err error) error {
				if !info.IsDir() && strings.Contains(info.Name(), ckecomponent.Spec.Lock) &&
					info.Name() != ckecomponent.Spec.Lock+"-"+ckecomponent.Name+".lock" {
					locked = true
				}
				return nil
			})
			if err != nil {
				return requeue, err
			}
			if locked {
				ckecomponent.Status.Reason = fmt.Sprintf("%s locked", ckecomponent.Spec.Lock)
				if err = r.Update(ctx, ckecomponent); err != nil {
					return requeue, err
				}
				requeue = true
				return requeue, err
			}
			lockFilePath := CkeComponentLockPath + "/" + ckecomponent.Spec.Lock + "-" + ckecomponent.Name + ".lock"
			if !IsFileExist(lockFilePath) {
				if _, err = os.Create(lockFilePath); err != nil {
					return requeue, err
				}
			}
		}

		if strings.Contains(ckecomponent.Status.Reason, "are not running") || strings.Contains(ckecomponent.Status.Reason, "locked") {
			ckecomponent.Status.Reason = ""
			ckecomponent.Status.Message = ""
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}

		var CkeJobs = &batchv1.JobList{}
		var JobSelector = map[string]string{
			"app":           ComponentInstallJobAppLabel,
			"componentName": ckecomponent.Name,
			"clusterID":     ckecluster.Name,
		}
		if err := r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
			return requeue, err
		}

		if len(CkeJobs.Items) == 0 {
			if err = r.Create(ctx, newComponentInstallJob(service.Spec.Ports[0].NodePort, ckecomponent, ckenodes, ckecluster)); err != nil {
				logrus.Errorf("Install Component, newComponentInstallJob error!")
				return requeue, err
			}
			logrus.Infof("success create job for install component(%s)", ckecomponent.Name)
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("Install Component, update component status error!")
				return requeue, err
			}
		} else {
			if JobCompleted(&CkeJobs.Items[0]) {
				if ckecomponent.Spec.Lock != "" {
					if err = CleanLockFile(ckecomponent); err != nil {
						return requeue, err
					}
				}
				if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
					logrus.Errorf("Install Component, delete job error!")
					return requeue, err
				}
				ckecomponent.Status.State = ComponentRunningStatus
				ckecomponent.Status.Version = ckecomponent.Spec.Version
				ckecomponent.Status.ImageTag = ckecomponent.Spec.ImageTag
				ckecomponent.Status.Parameters = ckecomponent.Spec.Parameters
				ckecomponent.Status.Nodes = nodes
				if err = r.Update(ctx, ckecomponent); err != nil {
					return requeue, err
				}
				return requeue, err
			}
			if JobFailed(&CkeJobs.Items[0]) {
				if ckecomponent.Spec.Lock != "" {
					if err = CleanLockFile(ckecomponent); err != nil {
						return requeue, err
					}
				}
				if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
					logrus.Errorf("Install Component, delete job error!")
					return requeue, err
				}
				ckecomponent.Status.State = ComponentInstallFailedStatus
				ckecomponent.Status.Reason = fmt.Sprintf("install component %s failed", ckecomponent.Name)
				ckecomponent.Status.Message = fmt.Sprintf("See log in /var/log/%s/%s/create for details.", ckecluster.Name, ckecomponent.Name)
				if err = r.Update(ctx, ckecomponent); err != nil {
					logrus.Errorf("Install Component, update component status error!")
					return requeue, err
				}
				return requeue, nil
			}
		}
	}

	requeue = true
	return requeue, err
}

func (r *CkeComponentReconciler) UpgradeComponent(ctx context.Context, service *corev1.Service, ckecomponent *ckev1alpha1.CkeComponent, ckecomponents *ckev1alpha1.CkeComponentList, ckenodes []ckev1alpha1.CkeNode, ckecluster *ckev1alpha1.CkeCluster) (bool, error) {
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, UpgradeComponent start")
	var requeue = false
	var err error

	if strings.Contains(ckecomponent.Spec.UnsupportAction, "upgrade") {
		ckecomponent.Status.Version = ckecomponent.Spec.Version
		ckecomponent.Status.ImageTag = ckecomponent.Spec.ImageTag
		ckecomponent.Status.Parameters = ckecomponent.Spec.Parameters
		if err = r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
		return requeue, err
	}

	if ckecomponent.Status.State != ComponentUpgradingStatus {
		ckecomponent.Status.State = ComponentUpgradingStatus
		ckecomponent.Status.Reason = ""
		ckecomponent.Status.Message = ""
		if err := r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
	}

	if !IsDependComponentRunning(ckecomponent, ckecomponents) {
		if ckecomponent.Status.Reason == "" {
			ckecomponent.Status.Reason = fmt.Sprintf("some depend ckecomponents are not running")
			ckecomponent.Status.Message = fmt.Sprintf("wait for depend ckecomponents running")
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}
	} else {
		if strings.Contains(ckecomponent.Status.Reason, "are not running") {
			ckecomponent.Status.Reason = ""
			ckecomponent.Status.Message = ""
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}

		var CkeJobs = &batchv1.JobList{}
		var JobSelector = map[string]string{
			"app":           ComponentUpgradeJobAppLabel,
			"componentName": ckecomponent.Name,
			"clusterID":     ckecluster.Name,
		}
		if err := r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
			return requeue, err
		}

		if len(CkeJobs.Items) == 0 {
			if err = r.Create(ctx, newComponentUpgradeJob(service.Spec.Ports[0].NodePort, ckecomponent, ckenodes, ckecluster)); err != nil {
				logrus.Errorf("Upgrade Component, newComponentUpgradeJob error!")
				return requeue, err
			}
			logrus.Infof("success create job for upgrade component(%s)", ckecomponent.Name)
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("Upgrade Component, update component status error!")
				return requeue, err
			}
		} else {
			if JobCompleted(&CkeJobs.Items[0]) {
				if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
					logrus.Errorf("Upgrade Component, delete job error!")
					return requeue, err
				}
				ckecomponent.Status.State = ComponentRunningStatus
				ckecomponent.Status.Version = ckecomponent.Spec.Version
				ckecomponent.Status.ImageTag = ckecomponent.Spec.ImageTag
				ckecomponent.Status.Parameters = ckecomponent.Spec.Parameters
				if err = r.Update(ctx, ckecomponent); err != nil {
					return requeue, err
				}
				return requeue, err
			}
			if JobFailed(&CkeJobs.Items[0]) {
				if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
					logrus.Errorf("Upgrade Component, delete job error!")
					return requeue, err
				}
				ckecomponent.Status.State = ComponentUpgradeFailedStatus
				ckecomponent.Spec.Version = ckecomponent.Status.Version
				ckecomponent.Spec.ImageTag = ckecomponent.Status.ImageTag
				ckecomponent.Spec.Parameters = ckecomponent.Status.Parameters
				ckecomponent.Status.Reason = fmt.Sprintf("upgrade component %s failed", ckecomponent.Name)
				ckecomponent.Status.Message = fmt.Sprintf("See log in /var/log/%s/%s/upgrade for details.", ckecluster.Name, ckecomponent.Name)
				if err = r.Update(ctx, ckecomponent); err != nil {
					logrus.Errorf("Upgrade Component, update component status error!")
					return requeue, err
				}
				return requeue, nil
			}
		}
	}
	requeue = true
	return requeue, err
}

func (r *CkeComponentReconciler) UpdateComponent(ctx context.Context, service *corev1.Service, ckecomponent *ckev1alpha1.CkeComponent, ckecomponents *ckev1alpha1.CkeComponentList, ckenodes []ckev1alpha1.CkeNode, ckecluster *ckev1alpha1.CkeCluster) (bool, error) {
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, UpdateComponent start")
	var requeue = false
	var err error

	if strings.Contains(ckecomponent.Spec.UnsupportAction, "update") {
		ckecomponent.Status.ImageTag = ckecomponent.Spec.ImageTag
		ckecomponent.Status.Parameters = ckecomponent.Spec.Parameters
		if err = r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
		return requeue, err
	}

	if ckecomponent.Status.State != ComponentUpdatingStatus {
		ckecomponent.Status.State = ComponentUpdatingStatus
		ckecomponent.Status.Reason = ""
		ckecomponent.Status.Message = ""
		if err := r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
	}

	if !IsDependComponentRunning(ckecomponent, ckecomponents) {
		if ckecomponent.Status.Reason == "" {
			ckecomponent.Status.Reason = fmt.Sprintf("some depend ckecomponents are not running")
			ckecomponent.Status.Message = fmt.Sprintf("wait for depend ckecomponents running")
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}
	} else {
		if strings.Contains(ckecomponent.Status.Reason, "are not running") {
			ckecomponent.Status.Reason = ""
			ckecomponent.Status.Message = ""
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}

		var CkeJobs = &batchv1.JobList{}
		var JobSelector = map[string]string{
			"app":           ComponentUpdateJobAppLabel,
			"componentName": ckecomponent.Name,
			"clusterID":     ckecluster.Name,
		}
		if err := r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
			return requeue, err
		}

		if len(CkeJobs.Items) == 0 {
			if err = r.Create(ctx, newComponentUpdateJob(service.Spec.Ports[0].NodePort, ckecomponent, ckenodes, ckecluster)); err != nil {
				logrus.Errorf("Update Component, newComponentUpdateJob error!")
				return requeue, err
			}
			logrus.Infof("success create job for update component(%s)", ckecomponent.Name)
			if err = r.Update(ctx, ckecomponent); err != nil {
				logrus.Errorf("Update Component, update component status error!")
				return requeue, err
			}
		} else {
			if JobCompleted(&CkeJobs.Items[0]) {
				if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
					logrus.Errorf("Update Component, delete job error!")
					return requeue, err
				}
				ckecomponent.Status.State = ComponentRunningStatus
				ckecomponent.Status.ImageTag = ckecomponent.Spec.ImageTag
				ckecomponent.Status.Parameters = ckecomponent.Spec.Parameters
				if err = r.Update(ctx, ckecomponent); err != nil {
					return requeue, err
				}
				return requeue, err
			}
			if JobFailed(&CkeJobs.Items[0]) {
				if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
					logrus.Errorf("Update Component, delete job error!")
					return requeue, err
				}
				ckecomponent.Status.State = ComponentUpdateFailedStatus
				ckecomponent.Spec.ImageTag = ckecomponent.Status.ImageTag
				ckecomponent.Spec.Parameters = ckecomponent.Status.Parameters
				ckecomponent.Status.Reason = fmt.Sprintf("update component %s failed", ckecomponent.Name)
				ckecomponent.Status.Message = fmt.Sprintf("See log in /var/log/%s/%s/update for details.", ckecluster.Name, ckecomponent.Name)
				if err = r.Update(ctx, ckecomponent); err != nil {
					logrus.Errorf("Update Component, update component status error!")
					return requeue, err
				}
				return requeue, nil
			}
		}
	}
	requeue = true
	return requeue, err
}

func (r *CkeComponentReconciler) ScaleupComponent(ctx context.Context, service *corev1.Service, ckecomponent *ckev1alpha1.CkeComponent, ckecomponents *ckev1alpha1.CkeComponentList, ckenodes []ckev1alpha1.CkeNode, ckecluster *ckev1alpha1.CkeCluster) (bool, error) {
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, ScaleupComponent start")
	var requeue = false
	var err error

	if strings.Contains(ckecomponent.Spec.UnsupportAction, "scaleup") {
		ckecomponent.Status.Nodes = AppendIfNotExist(ckecomponent.Status.Nodes, ckecomponent.Status.ScalingUpPending...)
		ckecomponent.Status.ScalingUpPending = nil
		if err = r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
		return requeue, err
	}

	if ckecomponent.Status.State != ComponentScalingUpStatus {
		ckecomponent.Status.State = ComponentScalingUpStatus
		ckecomponent.Status.Reason = ""
		ckecomponent.Status.Message = ""
		if err := r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
	}

	if !IsDependComponentRunning(ckecomponent, ckecomponents) {
		if ckecomponent.Status.Reason == "" {
			ckecomponent.Status.Reason = fmt.Sprintf("some depend ckecomponents are not running")
			ckecomponent.Status.Message = fmt.Sprintf("wait for depend ckecomponents running")
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}
	} else {
		if strings.Contains(ckecomponent.Status.Reason, "are not running") {
			ckecomponent.Status.Reason = ""
			ckecomponent.Status.Message = ""
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}

		for _, addNode := range ckecomponent.Status.ScalingUpPending {
			var CkeJobs = &batchv1.JobList{}
			var JobSelector = map[string]string{
				"app":           ComponentScaleupJobAppLabel,
				"componentName": ckecomponent.Name,
				"addNodeName":   addNode,
				"clusterID":     ckecluster.Name,
			}
			if err = r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
				return requeue, err
			}
			if len(CkeJobs.Items) == 0 {
				if err = r.Create(ctx, newComponentScaleUpJob(service.Spec.Ports[0].NodePort, ckecomponent, addNode, ckenodes, ckecluster)); err != nil {
					logrus.Errorf("Scaleup Component, newComponentScaleUpJob error!")
					return requeue, err
				}
				logrus.Infof("success create job for scaleup component(%s)", ckecomponent.Name)
			} else {
				if JobCompleted(&CkeJobs.Items[0]) {
					if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
						logrus.Errorf("Scaleup Component, delete job error!")
						return requeue, err
					}
					ckecomponent.Status.ScalingUpPending = RemoveStrings(ckecomponent.Status.ScalingUpPending, addNode)
					ckecomponent.Status.Nodes = AppendIfNotExist(ckecomponent.Status.Nodes, addNode)
					if strings.Contains(strings.Join(ckecomponent.Status.ScalingUpFailed, ","), addNode) {
						ckecomponent.Status.ScalingUpFailed = RemoveStrings(ckecomponent.Status.ScalingUpFailed, addNode)
					}
					if err = r.Update(ctx, ckecomponent); err != nil {
						return requeue, err
					}
				}
				if JobFailed(&CkeJobs.Items[0]) {
					if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
						logrus.Errorf("Scaleup Component, delete job error!")
						return requeue, err
					}
					if len(ckecomponent.Spec.Nodes) > 0 {
						ckecomponent.Spec.Nodes = RemoveStrings(ckecomponent.Status.Nodes, addNode)
					}
					ckecomponent.Status.ScalingUpPending = RemoveStrings(ckecomponent.Status.ScalingUpPending, addNode)
					ckecomponent.Status.ScalingUpFailed = AppendIfNotExist(ckecomponent.Status.ScalingUpFailed, addNode)
					if err = r.Update(ctx, ckecomponent); err != nil {
						return requeue, err
					}
				}
			}
		}

		if len(ckecomponent.Status.ScalingUpFailed) > 0 {
			ckecomponent.Status.Reason = fmt.Sprintf("scalingup component %s failed in nodes %s", ckecomponent.Name, strings.Join(ckecomponent.Status.ScalingUpFailed, ","))
			ckecomponent.Status.Message = fmt.Sprintf("See log in /var/log/%s/%s/scaleup for details.", ckecluster.Name, ckecomponent.Name)
			if err = r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}

		if len(ckecomponent.Status.ScalingUpPending) > 0 {
			requeue = true
		} else {
			ckecomponent.Status.State = ComponentRunningStatus
			if err := r.Update(ctx, ckecomponent); err != nil {
				return requeue, err
			}
		}
	}

	return requeue, err
}

func (r *CkeComponentReconciler) ScaledownComponent(ctx context.Context, service *corev1.Service, ckecomponent *ckev1alpha1.CkeComponent, ckenodes []ckev1alpha1.CkeNode, ckecluster *ckev1alpha1.CkeCluster) (bool, error) {
	logrus.Infof("CkeComponent Reconcile, CkeComponent=%s, ScaledownComponent start")
	var requeue = false
	var err error

	if strings.Contains(ckecomponent.Spec.UnsupportAction, "scaledown") {
		ckecomponent.Status.Nodes = RemoveStrings(ckecomponent.Status.Nodes, ckecomponent.Status.ScalingDownPending...)
		ckecomponent.Status.ScalingDownPending = nil
		if err = r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
		return requeue, err
	}

	if ckecomponent.Status.State != ComponentScalingDownStatus {
		ckecomponent.Status.State = ComponentScalingDownStatus
		ckecomponent.Status.Reason = ""
		ckecomponent.Status.Message = ""
		if err := r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
	}

	for _, deleteNode := range ckecomponent.Status.ScalingDownPending {
		var CkeJobs = &batchv1.JobList{}
		var JobSelector = map[string]string{
			"app":            ComponentScaledownJobAppLabel,
			"componentName":  ckecomponent.Name,
			"deleteNodeName": deleteNode,
			"clusterID":      ckecluster.Name,
		}
		if err = r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
			return requeue, err
		}
		if len(CkeJobs.Items) == 0 {
			if err = r.Create(ctx, newComponentScaleDownJob(service.Spec.Ports[0].NodePort, ckecomponent, deleteNode, ckenodes, ckecluster)); err != nil {
				logrus.Errorf("Scaledown Component, newComponentScaleDownJob error!")
				return requeue, err
			}
			logrus.Infof("success create job for scaledown component(%s)", ckecomponent.Name)
		} else {
			if JobCompleted(&CkeJobs.Items[0]) {
				if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
					logrus.Errorf("Scaledown Component, delete job error!")
					return requeue, err
				}
				ckecomponent.Status.ScalingDownPending = RemoveStrings(ckecomponent.Status.ScalingDownPending, deleteNode)
				ckecomponent.Status.Nodes = RemoveStrings(ckecomponent.Status.Nodes, deleteNode)
				if err = r.Update(ctx, ckecomponent); err != nil {
					return requeue, err
				}
			}
			if JobFailed(&CkeJobs.Items[0]) {
				if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
					logrus.Errorf("Scaledown Component, delete job error!")
					return requeue, err
				}
				ckecomponent.Status.ScalingDownPending = RemoveStrings(ckecomponent.Status.ScalingDownPending, deleteNode)
				ckecomponent.Status.ScalingDownFailed = AppendIfNotExist(ckecomponent.Status.ScalingDownFailed, deleteNode)
				if err = r.Update(ctx, ckecomponent); err != nil {
					return requeue, err
				}
			}
		}
	}

	if len(ckecomponent.Status.ScalingDownFailed) > 0 {
		ckecomponent.Status.Reason = fmt.Sprintf("scalingdown component %s failed in nodes %s", ckecomponent.Name, strings.Join(ckecomponent.Status.ScalingDownFailed, ","))
		ckecomponent.Status.Message = fmt.Sprintf("See log in /var/log/%s/%s/scaledown for details.", ckecluster.Name, ckecomponent.Name)
		if err = r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
	}

	if len(ckecomponent.Status.ScalingDownPending) > 0 {
		requeue = true
	} else {
		ckecomponent.Status.State = ComponentRunningStatus
		if err := r.Update(ctx, ckecomponent); err != nil {
			return requeue, err
		}
	}

	return requeue, err
}

func IsDependComponentRunning(component *ckev1alpha1.CkeComponent, ckecomponents *ckev1alpha1.CkeComponentList) bool {
	ckecomponentsMap := make(map[string]ckev1alpha1.CkeComponent)
	for i, _ := range ckecomponents.Items {
		ckecomponentsMap[ckecomponents.Items[i].Name] = ckecomponents.Items[i]
	}

	for _, depend := range component.Spec.Depend {
		if !strings.Contains(depend, "|") {
			if _, ok := ckecomponentsMap[depend]; !ok {
				return false
			}
			if ckecomponentsMap[depend].Status.State != ComponentRunningStatus && ckecomponentsMap[depend].Status.State != ComponentScalingUpStatus && ckecomponentsMap[depend].Status.State != ComponentScalingDownStatus {
				return false
			}
		} else {
			atLeastOneRunning := false
			d := strings.Split(depend, "|")
			for _, de := range d {
				if _, ok := ckecomponentsMap[de]; !ok {
					continue
				}
				if ckecomponentsMap[de].Status.State == ComponentRunningStatus && ckecomponentsMap[depend].Status.State != ComponentScalingUpStatus && ckecomponentsMap[depend].Status.State != ComponentScalingDownStatus {
					atLeastOneRunning = true
					break
				}
			}
			if !atLeastOneRunning {
				return false
			}
		}
	}

	return true
}

func IsFileExist(filename string) bool {
	var exist = true
	if _, err := os.Stat(filename); os.IsNotExist(err) {
		exist = false
	}
	return exist
}

func CleanLockFile(ckecomponent *ckev1alpha1.CkeComponent) error {
	lockFilePath := CkeComponentLockPath + "/" + ckecomponent.Spec.Lock + "-" + ckecomponent.Name + ".lock"
	err := os.RemoveAll(lockFilePath)
	if err != nil {
		return err

	}
	return err
}
